package cn.amossun.starter.event.listener;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import cn.amossun.starter.event.common.Constants;
import cn.amossun.starter.event.exception.EventMethodExecuteException;
import cn.amossun.starter.event.exception.EventMethodNodFoundException;
import cn.amossun.starter.event.property.EventContext;
import cn.amossun.starter.event.property.ListenerInvokeHandler;
import cn.amossun.starter.event.property.SimpleEventMessage;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import java.io.IOException;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

@Slf4j
public class SimpleRemoteEventMessageListener implements RemoteEventMessageListener<SimpleEventMessage>, ChannelAwareMessageListener {

    private final ThreadLocal<AcknowledgeMode> threadLocal = new ThreadLocal();

    private EventContext eventContext;

    public SimpleRemoteEventMessageListener(EventContext eventContext) {
        this.eventContext = eventContext;
    }

    public String[] getListeningQueues() {
        return this.eventContext.getQueueContexts().stream().map(queue -> {
            return queue.getQueueName();
        }).toArray(String[]::new);
    }

    @Override
    public void onMessage(Message message, Channel channel) {
        String queueName = message.getMessageProperties().getConsumerQueue();

        String messageStr = new String(message.getBody());

        SimpleEventMessage event = JSON.parseObject(messageStr, SimpleEventMessage.class);

        log.info("接收到来自队列: [{}] => {}", queueName, event);

        Boolean canAck = false;
        Boolean reQueue = false;

        try {
            onRemoteEvent(queueName, event);
        } catch (EventMethodNodFoundException e) {
            log.warn("未找到监听方法, 拒绝消费信息");
            canAck = true;
        } catch (EventMethodExecuteException e) {
            log.warn("调用监听方法异常, 拒绝消费信息");
            canAck = true; //TODO ExecuteException
            reQueue = e.isReTry();
        } catch (Exception e) {
            log.error("调用监听方法异常, 拒绝消费信息: ", e);
            canAck = true;
        }

        try {
            AcknowledgeMode acknowledgeMode = threadLocal.get();
            if(acknowledgeMode == null || acknowledgeMode.isAutoAck()) {
                return;
            }

            if(acknowledgeMode.isManual()) {
                /**
                 * {@link org.springframework.amqp.core.AcknowledgeMode}
                 */
                if (canAck) {
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, reQueue);
                    return;
                }
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                return;
            }

            if(canAck && acknowledgeMode == AcknowledgeMode.AUTO) {
                throw new EventMethodExecuteException("执行调用监听方法异常");
            }

        } catch (IOException e) {
            log.error("消费消息basicAck/basicNack异常: ", e);
            return;
        } finally {
            if(threadLocal.get() != null){
                threadLocal.remove();
            }
        }
    }

    @Override
    public void onRemoteEvent(String queueName, SimpleEventMessage event) {

        log.info("开始执行对应监听方法");

        try {
            String listenerKey = String.format(Constants.ROUTING_KEY_TEMPLATE, event.getTopic(), event.getDefinition());
            String allScopesKey = String.format(Constants.ROUTING_KEY_TEMPLATE, event.getTopic(), Constants.ROUTING_KEY_SUFFIX);

            List<ListenerInvokeHandler> listeners = Optional.ofNullable(eventContext.getInvokeHandlers(listenerKey)).orElseGet(() -> new ArrayList<>());
            List<ListenerInvokeHandler> allScopesListeners = Optional.ofNullable(eventContext.getInvokeHandlers(allScopesKey)).orElseGet(() -> new ArrayList<>());

            Set<ListenerInvokeHandler> listenerInvokeHandlers = new HashSet<>(listeners.size() + allScopesListeners.size());
            listenerInvokeHandlers.addAll(listeners);
            listenerInvokeHandlers.addAll(allScopesListeners);

            if (CollectionUtil.isEmpty(listenerInvokeHandlers)) {
                throw new EventMethodNodFoundException("监听方法未找到");
            }

            Map<List<String>, List<ListenerInvokeHandler>> queueNameMap = listenerInvokeHandlers.stream().collect(Collectors.groupingBy(ListenerInvokeHandler::getQueueNames));
            AtomicReference<List<ListenerInvokeHandler>> targetListeners = null;
            queueNameMap.entrySet().stream().forEach(queueNames->{
                if(queueNames.getKey().contains(queueName)) {
                    targetListeners.set(queueNames.getValue());
                }
            });

            if(CollectionUtil.isEmpty(targetListeners.get())) {
                throw new EventMethodNodFoundException("队列消费者未找到");
            }

            targetListeners.get().stream().forEach(listenerInvokeHandler -> {
                try {
                    AcknowledgeMode ackMode = listenerInvokeHandler.getListenerContext().getAckMode();
                    threadLocal.set(ackMode);

                    Object listenerInvokeBean = listenerInvokeHandler.getBean();

                    Method listenerInvokeMethod = listenerInvokeHandler.getMethod();

                    Class targetParamClass = listenerInvokeHandler.getListenerContext().getSchemaClass();

                    Object paramData = JSON.parseObject(event.getMessage(), targetParamClass);

                    listenerInvokeMethod.invoke(listenerInvokeBean, paramData);
                } catch (Exception ex) {
                    log.error("执行调用监听方法异常: ", ex);
                    throw new EventMethodExecuteException(ex.getMessage(), listenerInvokeHandler.getListenerContext().isReTry());
                }
            });

        } catch (EventMethodExecuteException ex) {
            throw ex;
        } catch (Exception ex) {
            throw new EventMethodExecuteException(ex.getMessage());
        }
    }
}
